import java.util.concurrent.*;

public class BrokerImple implements Broker{

    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);;
    int MaxQueueSize = 1000;

    //消息主题发过来
    //订阅者自己过来取

    ConcurrentHashMap<String,ArrayBlockingQueue<Consumer>> topicConsumers = new ConcurrentHashMap<>();
    //通过socket与Producer和Consumer连接
    @Override
    public void publish(Message message,long expirationTime) {
        ArrayBlockingQueue<Consumer> consumers = topicConsumers.get(message.getTopic());
        //对Topic设定10s过期时间，十秒后自动删除
        //10s后，消费者还没来取，就自动删除
        scheduler.schedule(()->{
            topicConsumers.remove(message.getTopic());
        },expirationTime, TimeUnit.SECONDS);
    }

    @Override
    public void subscribe(String topic, Consumer consumer) {
        topicConsumers.computeIfAbsent(topic,k-> new ArrayBlockingQueue<Consumer>(MaxQueueSize)).add(consumer);
    }

    @Override
    public void unsubscribe(String topic, Consumer consumer) {
        ArrayBlockingQueue<Consumer> consumers = topicConsumers.get(topic);
        if(consumers!=null){
            consumers.remove(consumer);
        }
    }






}
